package com.flink.cus.operators;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.util.Collector;

public class FMap1 implements FlatMapFunction<String, Tuple3<Long, Float, Long>>{

    @Override
    public void flatMap(String val_in, Collector<Tuple3<Long, Float, Long>> out) throws Exception {
        int timeidx = val_in.indexOf("/");
        if(timeidx<=1)
            return;//忽略这行输入
        Long time = 0L;
        try{
            time = Long.parseLong(val_in.substring(0, timeidx));
        }catch (Exception e){
            return;
        }        

        String rigthstr = val_in.substring(timeidx+1, val_in.length());
        String[] prisevols = rigthstr.split(";");
        for (String prsvol : prisevols) {
            String ary [] = prsvol.split("\\,");
            if(ary.length!=2) continue;
            String prs = ary[0];
            String vol = ary[1];
            if (prs.length() > 0 && vol.length()>0 ) {
                float dprs =0;
                long dvol =0; 
                try{
                    dprs = Float.parseFloat(prs);
                    dvol = Long.parseLong(vol);
                }catch(Exception e){
                    continue;
                } 
                if(dprs<=0 || dvol<0)
                    continue;
                Tuple3<Long, Float, Long> obj = new Tuple3<Long, Float, Long>(time, dprs, dvol);
                //System.out.println(obj);
                out.collect(obj);
            }
        }
    }
    
}
